package com.atguigu.realtime.dwd.db.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionCommentInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdInteractionCommentInfo().start(
            10012,
            4,
            "dwd_interaction_comment_info"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tableEnv, String groupID) {

        createTopicDb(groupID, tableEnv);

        createBasicDic(tableEnv);

        fitlerCommentInfo(tableEnv);

        Table joinTable = lookUpJoin(tableEnv);

        createKafkaSinkTable(tableEnv);

//        tableEnv.executeSql("select * from dwd_interaction_comment_info").print();

        joinTable.insertInto(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO).execute();
    }

    private static void createKafkaSinkTable(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table dwd_interaction_comment_info(" +
                " id STRING, \n" +
                " user_id STRING, \n" +
                " sku_id STRING, \n" +
                " spu_id STRING, \n" +
                " order_id STRING, \n" +
                " appraise_code STRING, \n" +
                " appraise_name STRING, \n" +
                " comment_txt STRING, \n" +
                " create_time STRING, \n" +
                " operate_time STRING" +
                ")" +
                SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
    }

    private static Table lookUpJoin(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select \n" +
                " `id`, \n" +
                " `user_id`, \n" +
//                " `nick_name`, \n" +
//                " `head_img`, \n" +
                " `sku_id`, \n" +
                " `spu_id`, \n" +
                " `order_id`, \n" +
                " `appraise`, \n" +
                " info.dic_name appraise_name,\n" +
                " `comment_txt`, \n" +
                " `create_time`, \n" +
                " `operate_time`\n" +
                "from comment_info c \n" +
                "join base_dic FOR SYSTEM_TIME AS OF c.proc_time as b\n" +
                "on c.appraise = b.rowkey");
    }

    private static void fitlerCommentInfo(StreamTableEnvironment tableEnv) {
        Table commentInfo = tableEnv.sqlQuery("select " +
                "`data`['id'] id, \n" +
                "`data`['user_id'] user_id, \n" +
                "`data`['nick_name'] nick_name, \n" +
                "`data`['head_img'] head_img, \n" +
                "`data`['sku_id'] sku_id, \n" +
                "`data`['spu_id'] spu_id, \n" +
                "`data`['order_id'] order_id, \n" +
                "`data`['appraise'] appraise, \n" +
                "`data`['comment_txt'] comment_txt, \n" +
                "`data`['create_time'] create_time, \n" +
                "`data`['operate_time'] operate_time, \n" +
                "proc_time \n" +
                "from topic_db where `database` = 'gmall' and `table` = 'comment_info' and `type` = 'insert'\n " +
                "");

        tableEnv.createTemporaryView("comment_info", commentInfo);
    }
}